package org.aurora.remote.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.aurora.common.serialization.Byte2ObjectSerializationFactory;
import org.aurora.common.util.OSUtil;
import org.aurora.remote.*;
import org.aurora.remote.exception.SendRequestFailException;
import org.aurora.remote.protocol.RemoteRequest;
import org.aurora.remote.protocol.RemoteResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class NettyRemoteServer implements RemoteServer<NettyEventSubscriber> {
    private static final Logger log = LoggerFactory.getLogger(NettyRemoteServer.class);
    private int port;

    private ServerBootstrap serverBootstrap;

    private EventLoopGroup bossEventLoopGroup;

    private EventLoopGroup workEventLoopGroup;

    private NettyRemoteCommandHandler nettyRemoteCommandHandler;

    private NettyConnectManager nettyConnectManager;

    public NettyRemoteServer(int port) {
        init(port);
    }

    private void init(int port) {
        this.port = port;

        boolean isLinux = OSUtil.isLinux();

        this.nettyRemoteCommandHandler = new NettyRemoteCommandHandler();

        this.nettyConnectManager = new NettyConnectManager();

        this.nettyRemoteCommandHandler.registerEventSubscriber(this.nettyConnectManager);

        this.bossEventLoopGroup = isLinux ? new EpollEventLoopGroup(1) : new NioEventLoopGroup(1);
        this.workEventLoopGroup = isLinux ? new EpollEventLoopGroup(3) : new NioEventLoopGroup(3);

        this.serverBootstrap = new ServerBootstrap();

        this.serverBootstrap
                .channel(isLinux ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .group(bossEventLoopGroup, workEventLoopGroup)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        ChannelPipeline pipeline = socketChannel.pipeline();

                        pipeline.addLast(NettyRemoteServer.this.nettyRemoteCommandHandler.getNettyConnectChangeHandler())
                                .addLast(NettyEncoder.getInstance())
                                .addLast(new NettyDecoder())
                                .addLast(nettyRemoteCommandHandler);
                    }
                });

    }

    @Override
    public void start() throws InterruptedException {
        this.serverBootstrap.bind(port).sync();
        this.nettyRemoteCommandHandler.start();
    }

    @Override
    public void shutdown() {
    }

    @Override
    public void registerProcessor(int commandCode, RemoteProcessor<RemoteRequest, RemoteResponse> processor) {
        this.nettyRemoteCommandHandler.registerProcessor(commandCode, processor);
    }

    @Override
    public void registerEventManager(NettyEventSubscriber nettyEventSubscriber) {
        this.nettyRemoteCommandHandler.registerEventSubscriber(nettyEventSubscriber);
    }

    @Override
    public <RESPONSE extends RemoteResponse> RESPONSE send(String addr, RemoteRequest request, long timeoutMillis, Class<RESPONSE> responseClazz) {
        Channel channel = this.nettyConnectManager.getChannel(addr);

        if (channel == null) {
            log.error("send message connect channel not exist, addr = {}", addr);
            throw new SendRequestFailException("send message connect channel not exist, addr = " + addr);
        }
        RemoteResponse remoteResponse = nettyRemoteCommandHandler.send(channel, request, timeoutMillis);

        return Byte2ObjectSerializationFactory.deserialization(remoteResponse.getData(), responseClazz, remoteResponse.getSerializationType());
    }

    @Override
    public <RESPONSE extends RemoteResponse> void sendAsync(String addr, RemoteRequest request, long timeoutMillis, SendCallback<RESPONSE> sendCallback) {
        Channel channel = this.nettyConnectManager.getChannel(addr);

        if (channel == null) {
            log.error("send sync message connect channel not exist, addr = {}", addr);
            throw new SendRequestFailException("send sync message connect channel not exist, addr = " + addr);
        }

        nettyRemoteCommandHandler.sendAsync(channel, request, timeoutMillis, sendCallback);
    }

    @Override
    public void disConnect(String addr) {
        this.nettyConnectManager.closeChannel(addr);
    }

    @Override
    public int listenerPort() {
        return port;
    }

}
